package com.ssm.mq.service;

import com.ssm.mq.config.AmqpConfig;
import lombok.RequiredArgsConstructor;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitMessagingTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.AlternativeJdkIdGenerator;
import org.springframework.util.IdGenerator;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

@Service
@RequiredArgsConstructor
public class AmqpProducer {
    final AmqpAdmin amqpAdmin;
    final RabbitTemplate rabbitTemplate;
    final RabbitMessagingTemplate rabbitMessagingTemplate;
    final IdGenerator idGenerator = new AlternativeJdkIdGenerator();

    @Scheduled(fixedDelay = 1000, initialDelay = 500)
    public void send() {
        rabbitTemplate.convertAndSend(AmqpConfig.EXCHANGE_NAME, AmqpConfig.ROUTING_KEY, UUID.randomUUID().toString(), message -> {
            message.getMessageProperties().setPriority(10); // 消息优先级，越大越高，会被优先消费
            return message;
        });
        rabbitTemplate.convertAndSend(AmqpConfig.QUEUE_NAME_TTL, idGenerator.generateId().toString());
        rabbitTemplate.convertAndSend(AmqpConfig.QUEUE_NAME_ANONYMOUS, idGenerator.generateId().toString());
        rabbitTemplate.convertAndSend(AmqpConfig.QUEUE_NAME_ANONYMOUS, "hello".getBytes(StandardCharsets.UTF_8));
        rabbitTemplate.convertAndSend(AmqpConfig.QUEUE_NAME_ANONYMOUS, 12345678);
        rabbitTemplate.convertAndSend("@RabbitListener", AmqpConfig.ROUTING_KEY, UUID.randomUUID().toString());
    }

    @Scheduled(fixedDelay = 1000, initialDelay = 500)
    public void sendTTL() {
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setExpiration("6000");
        messageProperties.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
        Message message = new Message("测试非持久化消息，且 6 秒过期，自动删除".getBytes(), messageProperties);
        rabbitTemplate.convertAndSend(AmqpConfig.QUEUE_NAME_TTL, message);
    }

    @Scheduled(fixedDelay = 1000, initialDelay = 500)
    public void sendError() {
        // 模拟消息发送到 EXCHANGE 失败
        CorrelationData correlationData = new CorrelationData(idGenerator.generateId().toString());
        rabbitTemplate.convertAndSend("error", AmqpConfig.ROUTING_KEY, UUID.randomUUID().toString(), correlationData);
        // 模拟消息发送到 EXCHANGE 成功，但 EXCHANGE 路由消息到 QUEUE 失败
        rabbitTemplate.convertAndSend(AmqpConfig.EXCHANGE_NAME, "error", UUID.randomUUID().toString(), correlationData);
    }
}